package com.kuizii.base.log;

import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.AppenderBase;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.Future;

/**
 * 演示写日志到kafka
 * 其中一种实现方法,可能会有bug，也可以采用{@link com.kuizii.base.log.KafkaAppender}
 */
public class KafkaLogAppender extends AppenderBase<ILoggingEvent> {
    private String bootstrapServers;
    private String topic;
    private KafkaProducer<String, String> producer;
    @Override
    public void start() {
        super.start();
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        props.put("max.block.ms",1000);
        props.put("batch.size", 16384);
        props.put("acks", "0");
        props.put("linger.ms", 100);
        props.put("buffer.memory", 33554432);
        producer = new KafkaProducer<>(props);
    }
    @Override
    public void stop() {
        super.stop();
        if (producer != null) {
            producer.close();
        }
    }
    @Override
    protected void append(ILoggingEvent eventObject) {
        String log = eventObject.getFormattedMessage();
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, log);
        producer.send(record);
//        Future<RecordMetadata> future = producer.send(record);
        // 如果需要同步等待发送结果，可以使用以下代码
        /*
        try {
            RecordMetadata metadata = future.get();
            // 处理metadata
        } catch (Exception e) {
            // 处理异常
        }
        */
    }
    public void setBootstrapServers(String bootstrapServers) {
        this.bootstrapServers = bootstrapServers;
    }
    public void setTopic(String topic) {
        this.topic = topic;
    }
}